package cn.wangjie.spark.report

import cn.wangjie.spark.config.ApplicationConfig
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

/**
 * 广告区域统计：ads_region_analysis，区域维度：省份和城市
 */
object AdsRegionAnalysisReport {
	
	/*
	不同业务报表统计分析时，两步骤：
	i. 编写SQL或者DSL分析
	ii. 将分析结果保存MySQL数据库表中
 */
	def doReport(dataframe: DataFrame): Unit = {
		
		// 获取SparkSession对象
		val spark: SparkSession = dataframe.sparkSession
		import spark.implicits._
		import org.apache.spark.sql.functions._
		
		val resultDF: DataFrame = dataframe
			// 第一步、按照维度分组：省份和城市
			.groupBy($"province", $"city")
			// 第二步、使用agg进行聚合操作, 主要使用CASE...wHEN...函数和SUM函数
			.agg(
				// 原始请求：requestmode = 1 and processnode >= 1
				sum(
					when($"requestmode".equalTo(1)
						.and($"processnode".geq(1)), 1
					).otherwise(0)
				).as("orginal_req_cnt"),
				// 有效请求：requestmode = 1 and processnode >= 2
				sum(
					when($"requestmode".equalTo(1)
						.and($"processnode".geq(2)), 1
					).otherwise(0)
				).as("valid_req_cnt"),
				// 广告请求：requestmode = 1 and processnode = 3
				sum(
					when($"requestmode".equalTo(1)
						.and($"processnode".equalTo(3)), 1
					).otherwise(0)
				).as("ad_req_cnt"),
				// 参与竞价数
				sum(
					when($"adplatformproviderid".geq(100000)
						.and($"iseffective".equalTo(1))
						.and($"isbilling".equalTo(1))
						.and($"isbid".equalTo(1))
						.and($"adorderid".notEqual(0)), 1
					).otherwise(0)
				).as("join_rtx_cnt"),
				// 竞价成功数
				sum(
					when($"adplatformproviderid".geq(100000)
						.and($"iseffective".equalTo(1))
						.and($"isbilling".equalTo(1))
						.and($"iswin".equalTo(1))
						.and($"adorderid".notEqual(0)), 1
					).otherwise(0)
				).as("success_rtx_cnt"),
				// 广告主展示数: requestmode = 2 and iseffective = 1
				sum(
					when($"requestmode".equalTo(2)
						.and($"iseffective".equalTo(1)), 1
					).otherwise(0)
				).as("ad_show_cnt"),
				// 广告主点击数: requestmode = 3 and iseffective = 1 and adorderid != 0
				sum(
					when($"requestmode".equalTo(3)
						.and($"iseffective".equalTo(1))
						.and($"adorderid".notEqual(0)), 1
					).otherwise(0)
				).as("ad_click_cnt"),
				// 媒介展示数
				sum(
					when($"requestmode".equalTo(2)
						.and($"iseffective".equalTo(1))
						.and($"isbilling".equalTo(1))
						.and($"isbid".equalTo(1))
						.and($"iswin".equalTo(1)), 1
					).otherwise(0)
				).as("media_show_cnt"),
				// 媒介点击数
				sum(
					when($"requestmode".equalTo(3)
						.and($"iseffective".equalTo(1))
						.and($"isbilling".equalTo(1))
						.and($"isbid".equalTo(1))
						.and($"iswin".equalTo(1)), 1
					).otherwise(0)
				).as("media_click_cnt"),
				// DSP 广告消费
				sum(
					when($"adplatformproviderid".geq(100000)
						.and($"iseffective".equalTo(1))
						.and($"isbilling".equalTo(1))
						.and($"iswin".equalTo(1))
						.and($"adorderid".gt(200000))
						.and($"adcreativeid".gt(200000)), floor($"winprice" / 1000)
					) otherwise (0)
				).as("dsp_pay_money"),
				// DSP广告成本
				sum(
					when($"adplatformproviderid".geq(100000)
						.and($"iseffective".equalTo(1))
						.and($"isbilling".equalTo(1))
						.and($"isbid".equalTo(1))
						.and($"iswin".equalTo(1))
						.and($"adorderid".gt(200000))
						.and($"adcreativeid".gt(200000)), floor($"adpayment" / 1000)
					) otherwise (0)
				).as("dsp_cost_money")
			)
			// 第三步、过滤非0数据
			.filter(
				$"join_rtx_cnt".notEqual(0)
					.and($"success_rtx_cnt".notEqual(0))
					.and($"ad_show_cnt".notEqual(0))
					.and($"ad_click_cnt".notEqual(0))
					.and($"media_show_cnt".notEqual(0))
					.and($"media_click_cnt".notEqual(0))
			)
			// 第四步、计算“三率”, 增加三列数据
			.withColumn(
				"success_rtx_rate", //
				round($"success_rtx_cnt" / $"join_rtx_cnt", 2) // 保留两位有效数字
			)
			.withColumn(
				"ad_click_rate", //
				round($"ad_click_cnt" / $"ad_show_cnt", 2) // 保留两位有效数字
			)
			.withColumn(
				"media_click_rate", //
				round($"media_click_cnt" / $"media_show_cnt", 2) // 保留两位有效数字
			)
			// 第五步、增加报表的日期
			.withColumn(
				"report_date", // 报表日期字段
				date_sub(current_date(), 1).cast(StringType)
			)
		
		// step3: 保存结果数据至MySQL表汇总
		saveResultToMySQL(resultDF)
	}
	
	
	/*
		不同业务报表统计分析时，两步骤：
		i. 编写SQL或者DSL分析
		ii. 将分析结果保存MySQL数据库表中
	 */
	def doReportWithAsSQL(dataframe: DataFrame): Unit = {
		
		// 获取SparkSession对象
		val spark: SparkSession = dataframe.sparkSession
		import spark.implicits._
		import org.apache.spark.sql.functions._
		
		// step1: 注册DataFrame为临时视图
		dataframe.createOrReplaceTempView("tmp_view_pmt")
		// step2: 编写SQL语句，并执行
		val resultDF: DataFrame = spark.sql(
			ReportSQLConstant.reportAdsRegionKpiSQL("tmp_view_pmt")
		)
		// step3: 保存结果数据至MySQL表汇总
		saveResultToMySQL(resultDF)
	}
	
	
	/*
		不同业务报表统计分析时，两步骤：
		i. 编写SQL或者DSL分析
		ii. 将分析结果保存MySQL数据库表中
	 */
	def doReportSQL(dataframe: DataFrame): Unit = {
		
		// 获取SparkSession对象
		val spark: SparkSession = dataframe.sparkSession
		import spark.implicits._
		import org.apache.spark.sql.functions._
		
		// step1: 注册DataFrame为临时视图
		dataframe.createOrReplaceTempView("tmp_view_pmt")
		// step2: 编写SQL语句，并执行
		val reportDF: DataFrame = spark.sql(
			ReportSQLConstant.reportAdsRegionSQL("tmp_view_pmt")
		)
		//reportDF.show(20, truncate = false)
		
		// step3：计算“三率”，继续注册临时视图，编写SQL
		reportDF.createOrReplaceTempView("tmp_view_report")
		val resultDF: DataFrame = spark.sql(
			ReportSQLConstant.reportAdsRegionRateSQL("tmp_view_report")
		)
	
		// step4: 保存结果数据至MySQL表汇总
		saveResultToMySQL(resultDF)
	}
	
	
	/**
	 * 将分析报表数据DataFrame保存至MySQL表中
	 */
	def saveResultToMySQL(dataframe: DataFrame): Unit = {
		dataframe
			.coalesce(1)
			.write
			.mode(SaveMode.Append)
			.format("jdbc")
			.option("driver", ApplicationConfig.MYSQL_JDBC_DRIVER)
			.option("url", ApplicationConfig.MYSQL_JDBC_URL)
			.option("user", ApplicationConfig.MYSQL_JDBC_USERNAME)
			.option("password", ApplicationConfig.MYSQL_JDBC_PASSWORD)
			.option("dbtable", "itcast_ads_report.ads_region_analysis")
			.save()
	}
}
